package com.zhaorujie.rpc.core.remoting.netty.server;

import com.zhaorujie.rpc.core.remoting.netty.codec.NettyDecoder;
import com.zhaorujie.rpc.core.remoting.netty.codec.NettyEncoder;
import com.zhaorujie.rpc.core.remoting.params.Beat;
import com.zhaorujie.rpc.core.remoting.params.RpcRequest;
import com.zhaorujie.rpc.core.remoting.params.RpcResponse;
import com.zhaorujie.rpc.core.remoting.provider.RpcProviderFactory;
import com.zhaorujie.rpc.core.utils.ThreadPoolUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author 赵汝杰
 * @version v1.0
 * @description NettyServer
 * @date 2024/3/5
 **/
public class NettyServer {

    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);

    private final RpcProviderFactory rpcProviderFactory;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private ThreadPoolExecutor serviceThreadPool;

    public NettyServer(RpcProviderFactory rpcProviderFactory) {
        this.rpcProviderFactory = rpcProviderFactory;
    }

    /**
     * 启动netty服务端
     */
    public void start() {
        // 单独启用一个线程来处理netty服务启动的一些资源配置
        new Thread(() -> {
            // 创建一个业务线程单独处理调用业务，反正io线程被占用
            this.serviceThreadPool = ThreadPoolUtil.createServiceThreadPool(rpcProviderFactory.getCorePoolSize(), rpcProviderFactory.getMaxPoolSize());

            this.bossGroup = new NioEventLoopGroup();
            this.workerGroup = new NioEventLoopGroup();

            ServerBootstrap bootstrap = new ServerBootstrap();

            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childOption(ChannelOption.TCP_NODELAY, false)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            channel.pipeline()
                                    // 日志打印
                                    .addLast(new LoggingHandler(LogLevel.DEBUG))
                                    // 添加心跳读写空闲处理
                                    .addLast(new IdleStateHandler(0, 0, Beat.BEAT_TIMEOUT * 3))
                                    .addLast("nettyDecoder", new NettyDecoder(RpcRequest.class, rpcProviderFactory.getSerializer()))
                                    .addLast("nettyEncoder", new NettyEncoder(RpcResponse.class, rpcProviderFactory.getSerializer()))
                                    .addLast("nettyServerHandler", new NettyServerHandler(rpcProviderFactory, serviceThreadPool));
                        }
                    });

            try {
                // channel绑定
                ChannelFuture channelFuture = bootstrap.bind(rpcProviderFactory.getPort());
                // 通过回调的方式判断是启动成功还是失败
                channelFuture.addListener(future -> {
                    if (future.isSuccess()) {
                        logger.info("rpc remoting server start success");
                    } else {
                        logger.error("rpc remoting server start fail");
                    }
                });

                // channel关闭
                channelFuture.channel().closeFuture().addListener(future -> {
                    if (future.isSuccess()) {
                        logger.info("rpc remoting server destroy success");
                    } else {
                        logger.error("rpc remoting server destroy fail");
                    }
                });

            } catch (Exception e) {
                logger.error("rpc remoting server error.", e);
            }
        }).start();
    }

    /**
     * 注册一个关闭的钩子函数
     */
    public void registerShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            if (serviceThreadPool != null) {
                // 销毁线程池
                serviceThreadPool.shutdown();
            }
            if (bossGroup != null) {
                bossGroup.shutdownGracefully();
            }
            if (workerGroup != null) {
                workerGroup.shutdownGracefully();
            }
            logger.info("rpc remoting server destroy success.");
        }));
    }
}
